package seven.five;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

public class MultipleOutputTest {
	static class MyMapper extends Mapper<LongWritable,Text,Text,Text>{

		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
			String[] fiels = line.split(",");
			context.write(new Text(fiels[0]),value);
		} 
	}
	
	static class MyReducer extends Reducer<Text,Text,NullWritable,Text>{
		private MultipleOutputs<NullWritable,Text> mulitpleOutputs;
		@Override
		protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, NullWritable, Text>.Context context)
				throws IOException, InterruptedException {
			for(Text value : values){
				mulitpleOutputs.write(NullWritable.get(), value, key.toString());
			}
			
		}

		@Override
		protected void cleanup(Reducer<Text, Text, NullWritable, Text>.Context context)
				throws IOException, InterruptedException {
			mulitpleOutputs.close();
		}

		@Override
		protected void setup(Reducer<Text, Text, NullWritable, Text>.Context context)
				throws IOException, InterruptedException {
			mulitpleOutputs = new MultipleOutputs<NullWritable,Text>(context); 
		} 
	}
	public static void main(String[] args)throws Exception {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);
		
		job.setJarByClass(MultipleOutputTest.class);
		
		job.setMapperClass(MyMapper.class);
		job.setReducerClass(MyReducer.class);
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		
		job.setOutputKeyClass(NullWritable.class);
		job.setOutputValueClass(Text.class);
		
		
		FileInputFormat.setInputPaths(job,new Path("E:/code/workspace/hadoopone/src/main/java/seven/entity/input/"));
		//指定job输出结果目录所在
		FileOutputFormat.setOutputPath(job, new Path("E:/code/workspace/hadoopone/src/main/java/seven/entity/output/"));
		
		
		job.waitForCompletion(true);
		
	}
}
